package com.yupi.springbootinit.service;

import com.alibaba.fastjson.JSON;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry.*;
import com.alibaba.otter.canal.protocol.Message;
import com.yupi.springbootinit.model.dto.post.PostEsDTO;
import com.yupi.springbootinit.model.entity.Post;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.CommandLineRunner;
import org.springframework.data.elasticsearch.core.ElasticsearchRestTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.net.InetSocketAddress;
import java.util.Date;
import java.util.List;

@Slf4j
@Component
public class CanalToEsSyncService implements CommandLineRunner {

    @Resource
    private ElasticsearchRestTemplate elasticsearchRestTemplate;

    @Override
    public void run(String... args) {
        startCanalClient();
    }

    private void startCanalClient() {
        CanalConnector connector = CanalConnectors.newSingleConnector(
                new InetSocketAddress("127.0.0.1", 11111),
                "example", "", "");

        int batchSize = 1000;
        try {
            connector.connect();
            connector.subscribe("yuso.post"); // 只订阅特定库表

            while (true) {
                Message message = connector.getWithoutAck(batchSize);
                long batchId = message.getId();
                if (batchId == -1 || message.getEntries().isEmpty()) {
                    Thread.sleep(1000);
                    continue;
                }

                try {
                    processEntries(message.getEntries());
                    connector.ack(batchId);
                } catch (Exception e) {
                    log.error("处理canal消息失败", e);
                    connector.rollback(batchId);
                }
            }
        } catch (Exception e) {
            log.error("Canal客户端异常", e);
        } finally {
            connector.disconnect();
        }
    }

    private void processEntries(List<Entry> entries) {
        for (Entry entry : entries) {
            if (entry.getEntryType() != EntryType.ROWDATA) {
                continue;
            }

            try {
                RowChange rowChange = RowChange.parseFrom(entry.getStoreValue());
                String schema = entry.getHeader().getSchemaName();
                String table = entry.getHeader().getTableName();

                if (!"yuso".equals(schema) || !"post".equals(table)) {
                    continue;
                }

                for (RowData rowData : rowChange.getRowDatasList()) {
                    switch (rowChange.getEventType()) {
                        case INSERT:
                            handleInsert(rowData);
                            break;
                        case UPDATE:
                            handleUpdate(rowData);
                            break;
                        case DELETE:
                            handleDelete(rowData);
                            break;
                    }
                }
            } catch (Exception e) {
                log.error("解析binlog失败", e);
            }
        }
    }


    // 新增从Canal行数据直接转Post对象的方法
    private Post convertToPost(List<Column> columns) {
        Post post = new Post();
        for (Column column : columns) {
            String columnName = column.getName();
            String value = column.getValue();

            switch (columnName) {
                case "id":
                    post.setId(Long.parseLong(value));
                    break;
                case "title":
                    post.setTitle(value);
                    break;
                case "content":
                    post.setContent(value);
                    break;
                case "tags":
                    post.setTags(JSON.parseArray(value, String.class).toString()); // 假设tags是JSON数组
                    break;
                case "user_id":
                    post.setUserId(Long.parseLong(value));
                    break;
                case "create_time":
                    post.setCreateTime(new Date(Long.parseLong(value)));
                    break;
                case "update_time":
                    post.setUpdateTime(new Date(Long.parseLong(value)));
                    break;
                case "is_delete":
                    post.setIsDelete(Integer.parseInt(value));
                    break;
                default:
                    log.warn("未知列: {}", columnName);
            }
        }
        return post;
    }

    // 修改handleInsert和handleUpdate方法
    private void handleInsert(RowData rowData) {
        Post post = convertToPost(rowData.getAfterColumnsList());
        PostEsDTO postEsDTO = PostEsDTO.objToDto(post);
        elasticsearchRestTemplate.save(postEsDTO);
        log.info("新增ES文档: {}", postEsDTO.getId());
    }

    private void handleUpdate(RowData rowData) {
        Post post = convertToPost(rowData.getAfterColumnsList());
        PostEsDTO postEsDTO = PostEsDTO.objToDto(post);
        elasticsearchRestTemplate.save(postEsDTO);
        log.info("更新ES文档: {}", postEsDTO.getId());
    }

    private void handleDelete(RowData rowData) {
        String id = rowData.getBeforeColumnsList().get(0).getValue();
        elasticsearchRestTemplate.delete(id, PostEsDTO.class);
        log.info("删除ES文档: {}", id);
    }

}
